package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSqlApp;
import com.atguigu.gmall.realtime.bean.KeyWordStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.IkAnalyzer;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/11/3 14:30
 */
public class DwsSearchKeyWordApp extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsSearchKeyWordApp().init(4004, 1, "DwsSearchKeyWordApp");
    }
    
    @Override
    protected void run(StreamTableEnvironment tEnv) {
        // 1. 建动态表与 dwd_page进行关联
        tEnv.executeSql("create table page(" +
                            "   page map<string, string>, " +
                            "   ts bigint," +
                            "   et as to_timestamp(from_unixtime(ts/1000)), " +
                            "   watermark for et as et - interval '3' second " +
                            ")with(" +
                            "   'connector'='kafka', " +
                            "   'properties.bootstrap.servers'='hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                            "   'properties.group.id'='DwsSearchKeyWordApp', " +
                            "   'topic'='" + Constant.TOPIC_DWD_PAGE + "', " +
                            "   'scan.startup.mode'='latest-offset',  " +
                            "   'format'='json'  " +
                            ")");
        // tEnv.sqlQuery("select * from page").execute().print();
        // 2. 过滤出现需要的数据
        Table t1 = tEnv.sqlQuery("select" +
                                     " page['item'] keyword, " +
                                     " et " +
                                     "from page " +
                                     "where page['page_id'] = 'good_list' " +
                                     "and page['item'] is not null " +
                                     "and page['item_type'] = 'keyword'");
        tEnv.createTemporaryView("t1", t1);
        // 3. 对关键词进行分词
        // 自定义函数: 1. 标量函数 2. 表值函数 3. 聚合函数  4. 表值聚合
        // 3.1 注册自定义函数
        tEnv.createTemporaryFunction("ik_analyzer", IkAnalyzer.class);
        // 3.2 sql中使用
        Table t2 = tEnv.sqlQuery("select" +
                                     " word, " +
                                     " et " +
                                     "from t1 " +
                                     "join lateral table(ik_analyzer(keyword)) on true");
        tEnv.createTemporaryView("t2", t2);
        
        // 4. 开窗聚合
        Table result = tEnv.sqlQuery("select" +
                                         " date_format(tumble_start(et, interval '5' second), 'yyyy-MM-dd HH:mm:ss') stt, " +
                                         " date_format(tumble_end(et, interval '5' second), 'yyyy-MM-dd HH:mm:ss') edt," +
                                         " word keyword, " +
                                         " 'search' source, " +
                                         " count(word) ct," +  // sum(1) count(*) count(word)
                                         " unix_timestamp() * 1000 ts " +
                                         "from t2 " +
                                         "group by " +
                                         " word, " +
                                         " tumble(et, interval '5' second)");
        // 5. 写出到ClickHouse中
        tEnv
            .toRetractStream(result, KeyWordStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink("gmall2021", "keyword_stats_2021", KeyWordStats.class));
        
    }
}
